#include <algorithm>
#include <functional>
#include "port.h"
#include "asynevent.h"
#include "message.h"

#ifdef WIN32
#include <Ws2ipdef.h>
#endif

#define TEMP_BUFFER 4096
//#define PROBING_DELAY 100 //100ms
#define PRECISION_TICK 10 //10ms

#define PROBING_MANAGER


//////////////////////////////////////////////////////////////////////////
//    Port
//////////////////////////////////////////////////////////////////////////

Port::~Port()
{
	evutil_closesocket(socket_);

	//Remove all connections
	AddressMap tmp = connections_;
	AddressMap::iterator iter = tmp.begin();
	AddressMap::iterator end = tmp.end();
	for(; iter!=end; ++iter){
		delete iter->second;
	}
}

void Port::AddConnection(Connection* conn) {
	connections_[conn->RemoteAddress()] = conn;
	OnCreated(conn);
}

void Port::RemoveConnection(Connection* conn) 
{
	AddressMap::iterator iter = connections_.find(conn->RemoteAddress());
	assert(iter != connections_.end());
	connections_.erase(iter);
	OnDestroy(conn);
}

Connection* Port::GetConnection(const SocketAddress& remote_addr) 
{
	AddressMap::const_iterator iter = connections_.find(remote_addr);
	if (iter != connections_.end())
		return iter->second;
	else
		return NULL;
}

void Port::OnCreated(IConnection* pConn)
{
	std::for_each(listener_set_.begin(), listener_set_.end(), std::bind2nd(std::mem_fun1(&IConnListener::OnCreated), pConn));
}

void Port::OnConnected(IConnection* pConn)
{
	std::for_each(listener_set_.begin(), listener_set_.end(), std::bind2nd(std::mem_fun1(&IConnListener::OnConnected), pConn));
}

void Port::OnReadPacket(IConnection* pConn, const char* data, size_t size)
{
	std::set<IConnListener *>::iterator iter = listener_set_.begin();
	std::set<IConnListener *>::iterator end = listener_set_.end();
	for(; iter!=end; ++iter){
		(*iter)->OnReadPacket(pConn, data, size);
	}
}

void Port::OnWritable(IConnection* pConn, bool writable)
{
	std::set<IConnListener *>::iterator iter = listener_set_.begin();
	std::set<IConnListener *>::iterator end = listener_set_.end();
	for(; iter!=end; ++iter){
		(*iter)->OnWritable(pConn, writable);
	}
}

void Port::OnClose(IConnection* pConn, int err)
{
	std::set<IConnListener *>::iterator iter = listener_set_.begin();
	std::set<IConnListener *>::iterator end = listener_set_.end();
	for(; iter!=end; ++iter){
		(*iter)->OnClose(pConn, err);
	}
}

void Port::OnError(IConnection* pConn, int err)
{
	std::set<IConnListener *>::iterator iter = listener_set_.begin();
	std::set<IConnListener *>::iterator end = listener_set_.end();
	for(; iter!=end; ++iter){
		(*iter)->OnError(pConn, err);
	}

	//should delete here?
	//delete pConn;
	
	//It's difficult to delete here, if there're more than one OnError, it will delete multiple times
	//So, leave it to the App
	//Dispose(pConn);
}

void Port::OnDestroy(IConnection* pConn)
{
	std::for_each(listener_set_.begin(), listener_set_.end(), std::bind2nd(std::mem_fun1(&IConnListener::OnDestroy), pConn));
}


//void Port::OnReadPacket(const char* data, size_t size, const SocketAddress& addr)
//{
//	Connection* conn = CreateConnection(addr);
//	if (conn) {
//		conn->OnReadPacket(data, size);
//	} 
//}

int SetOption(evutil_socket_t s, Option opt, int value) 
{
	if(opt == OPT_DONTFRAGMENT){
#ifdef WIN32
		value = (value == 0) ? 0 : 1;
		return ::setsockopt(s, IPPROTO_IP, IP_DONTFRAGMENT, 
			                reinterpret_cast<char*>(&value), sizeof(value));
#endif
#ifdef __linux 
		value = (value == 0) ? IP_PMTUDISC_DONT : IP_PMTUDISC_DO;
		return ::setsockopt(s, IPPROTO_IP, IP_MTU_DISCOVER, 
			                &value, sizeof(value));
#endif
#ifdef OSX
		// This is not possible on OSX.
		return -1;
#endif
	}
	else if(opt == OPT_REUSEADDR){
		value = (value == 0) ? 0 : 1;
		return ::setsockopt( s, SOL_SOCKET, SO_REUSEADDR, (const char*)&value, sizeof(value) );
	}
	else if(opt == OPT_RCVBUF){
		return  ::setsockopt( s, SOL_SOCKET, SO_RCVBUF, (const char *)&value, sizeof(value) );
	}
	else if(opt == OPT_NODELAY){
		return  ::setsockopt( s, IPPROTO_TCP, TCP_NODELAY, (const char *)&value,sizeof(value));
	}

	return -1;
}

//////////////////////////////////////////////////////////////////////////
//    Connection
//////////////////////////////////////////////////////////////////////////

Connection::Connection(Port* port, const SocketAddress& remote_addr)
:port_(port)
,remote_addr_(remote_addr)
,connected_(false)
,recv_total_bytes_(0)
,sent_total_bytes_(0)
,owd_base_(0)
,last_io_ts_(0)
,start_ts_(0)
,life_time_(0)
,seq_expt_(0)
,seq_init_(0)
,pkt_loss_(0)
,interval_(PROBING_DELAY)
,ping_enabled_(true)
,error_(0)
,smoothed_rtt_(0)
{
	//use a random id to avoid confusing of disorder pkts
	conv_id_ = RANDOM_16();
	ptk_seq_ = RANDOM_32();
	last_ping_sent_ = Now_ms();

	evbuf_output_ = evbuffer_new();
	evbuf_input_ = evbuffer_new();

	probe_timer_.AddListener(this);
}

Connection::~Connection()
{
	probe_timer_.Stop();
	probe_timer_.RemoveListener(this);

	port_->RemoveConnection(this);

	evbuffer_free(evbuf_output_);
	evbuffer_free(evbuf_input_);

	for(std::deque<NtdProto*>::iterator iter = pdu_queue_.begin();
		iter < pdu_queue_.end(); ++iter)
	{
		delete *iter;
	}
}

void Connection::EnablePing(bool enabled) 
{ 
	ping_enabled_ = enabled; 
#ifndef PROBING_MANAGER
	if( enabled ){
		SetProbingInterval(PROBING_DELAY);
	}else{
		SetProbingInterval(INT_MAX);
	}
#endif
}

void Connection::SetProbingInterval(int ms)
{
#ifdef PROBING_MANAGER
	interval_ = ms;
#else
	if( ms == INT_MAX ){
		//stop probing timer
		interval_ = ms;
		probe_timer_.Stop();
	}else{
		// if( ms != interval_ )
		interval_ = ms;
		probe_timer_.Stop();
		if( connected_ ){
			probe_timer_.Start(interval_, true);
		}
	}
#endif
}

bool Connection::Flush()
{
	//sent data in the output buffer and the PDU queue
	while( HasRemainPDU() ){
		int32 ret = SendOutputbuf();
		if( ret <= 0 ){
			return false;
		}
	}

	//go on deal with the queue
	while( !pdu_queue_.empty() ){
		NtdProto * proto = pdu_queue_.front();
		pdu_queue_.pop_front();
		bool ret = TrySend(proto);

		if( !ret ){
			return false;
		}
	}

	return true;
}

int32 Connection::SendOutputbuf()
{
	//send packets
	int8 packet[TEMP_BUFFER];
	int32 length = evbuffer_get_length(evbuf_output_);

	if( length > 0 ){
		evbuffer_copyout(evbuf_output_, packet, length);

		int32 nSendBytes = DoSend(packet, length);

		//TCP may sent partial data, we have to cope with it by resent it in the next write event
		if( nSendBytes > 0 ){
			evbuffer_drain(evbuf_output_, nSendBytes);
			if( nSendBytes < length){
				LOG4CPLUS_DEBUG(g_logger,"Not all sent, sent them in the next write event");
			}
		}

		return nSendBytes;
	}else{
		return 0;
	}
}

bool Connection::HasRemainPDU()
{
	return evbuffer_get_length(evbuf_output_) != 0;
}

int Connection::Send(byte * data, size_t size)
{
	if( !HasRemainPDU() ){

		if( ping_enabled_ ){
			int32 now = Now_ms();
			NtdProto * proto = NULL;

			bool can_ping = IsTimetoPing(now);
			if( can_ping ){
				LOG4CPLUS_THIS_TRACE(g_logger, "ping using appdata--->");
				NtdProtoData * proto_data = new NtdProtoData();
				proto_data->m_pData = data;
				proto_data->m_len = size;
				proto_data->m_ptk_seq = NtdSeq::incseq(ptk_seq_);
				proto_data->m_timestamp_sending = now;

				//Only increase when sent succeed
				//ptk_seq_ = proto_data->m_ptk_seq;
				//last_ping_sent_ = now;

				proto = proto_data;
			}else{
				NtdProtoRawData * proto_rawdata = new NtdProtoRawData();
				proto_rawdata->m_pData = data;
				proto_rawdata->m_len = size;

				proto = proto_rawdata;
			}

			proto->m_convId = conv_id_;
			proto->Encode(evbuf_output_);

			int32 nSendBytes = SendOutputbuf();

			//TCP may sent partial data, we have to cope with it by resent it in the next write event
			if( nSendBytes > 0 ){
				if(can_ping){
					ptk_seq_ = NtdSeq::incseq(ptk_seq_);
					last_ping_sent_ = now;

#ifdef PROBING_MANAGER
					ProbingManager::Instance()->OnTick();
#endif
				}
				nSendBytes = size;
				//LOG4CPLUS_TRACE(g_logger, proto->Desc());
				//LOG4CPLUS_DEBUG(g_logger,"now = "<<ping->m_timestamp_sending);
			}

			delete proto;
			return nSendBytes;
		}else{
			//just send it out
			return DoSend(data, size);
		}

	}else{
		return -1;
	}
}

bool Connection::TrySend(NtdProto* proto, bool urgent)
{
	if( !HasRemainPDU() ){
		proto->Encode(evbuf_output_);

		int32 nSendBytes = SendOutputbuf();

		//TCP may sent partial data, we have to cope with it by resent it in the next write event
		if( nSendBytes > 0 ){			
			//LOG4CPLUS_TRACE(g_logger, proto->Desc());
			//LOG4CPLUS_DEBUG(g_logger,"now = "<<ping->m_timestamp_sending);
		}
		delete proto;

		if( HasRemainPDU() ){
			return false;
		}else{
			return true;
		}
	}else{
		if( urgent ){
			pdu_queue_.push_front(proto);
		}else{
			pdu_queue_.push_back(proto);
		}
		return false;
	}
}

bool Connection::IsTimetoPing(uint32 now)
{
	if( last_ping_sent_ + interval_ > now ){
		//LOG4CPLUS_TRACE(g_logger, "We've already pinged");
		return false;
	}else{
		return true;
	}
}

void Connection::DoPing()
{
	LOG4CPLUS_THIS_TRACE(g_logger, "ping--->");

	uint32 now = Now_ms();

	NtdProtoData * ping = new NtdProtoData;
	//Send probing packets now
	ping->m_convId = conv_id_;
	ping->m_ptk_seq = NtdSeq::incseq(ptk_seq_);
	ping->m_timestamp_sending = now;
	ptk_seq_ = ping->m_ptk_seq;
	last_ping_sent_ = now;

	//Make the ping pkts send at nearly the same time in order to find correlation-ship
	TrySend(ping, true);
}

//ONLY PDU NtdProtoData and PDU NtdProtoFeedback will be used now
void Connection::OnTimer(NTDTimer* pTimer)
{
	//TODO, send a fleet or just one? Need to avoid clock skew problems

	DoPing();
	//uint32 now = Now_ms();
	//if( IsTimetoPing(now) && connected_ ){
	//	LOG4CPLUS_TRACE(g_logger, "ping--->");
	//	NtdProtoData * ping = new NtdProtoData;
	//	//Send probing packets now
	//	ping->m_convId = conv_id_;
	//	ping->m_ptk_seq = NtdSeq::incseq(ptk_seq_);
	//	ping->m_timestamp_sending = now;
	//	ptk_seq_ = ping->m_ptk_seq;
	//	last_ping_sent_ = now;
	//	
	//	TrySend(ping);
	////}else{
	////	LOG4CPLUS_TRACE(g_logger, "We've already pinged by APP data");
	//}

}

void Connection::SetConnected(bool value) 
{
	bool old_value = connected_;
	connected_ = value;
	if (value != old_value) {
		LOG4CPLUS_DEBUG(g_logger, "set_connected: "<<value);

#ifndef PROBING_MANAGER
		if( value && ping_enabled_ ){
			probe_timer_.Start(interval_, true);
		}else{
			probe_timer_.Stop();
		}
#endif
	}
}

//here we only cope with packets forward to the APP layer
void Connection::OnReadPacket(const char* data, size_t size)
{
	recv_total_bytes_ += size;

	if( ping_enabled_ ){
		//decode first, handle probing packets and decode the rawData PDU.
		//Try decode first, if not underlying protocol, forward to the APP layer
		//If data with non-zero length, decode the data out and forward to the APP layer
		//Send feedbacks to the other side

		evbuffer_add(evbuf_input_, data, size);

		//sockaddr_in saddr;
		//remote_addr_.ToSockAddr(&saddr);

		while(true){
			bool ret = NtdProtoParser::ParseProtocol(evbuf_input_, NULL);
			if( ret == false ){
				break;
			}
		}
	}else{
		//forward to APP layer
		port_->OnReadPacket(this, data, size);
	}
}

void Connection::AddSample(const NtdMetrics& metrics)
{
	std::for_each(monitor_set_.begin(), monitor_set_.end(), std::bind2nd(std::mem_fun1(&IConnMonitor::AddSample), &metrics));
}

void Connection::HandleProtoData(NtdProtoData * proto, void* pContext)
{
	//signal read event
	port_->OnReadPacket(this, (const char*)proto->m_pData, proto->m_len);

	//send back a feedback response
	uint32 ts = Now_ms();

	last_io_ts_ = ts;
	if( start_ts_ == 0 ){
		start_ts_ = last_io_ts_;
		seq_init_ = proto->m_ptk_seq;
		seq_expt_ = seq_init_;
	}

	//m_sm.Add(proto->GetPaketSize());
	//recv_total_bytes_ += proto->GetPaketSize();
	life_time_ = last_io_ts_ - start_ts_;

	LOG4CPLUS_TRACE(g_logger," [server] - "<< proto->Desc());

	//pkt loss
	if( NtdSeq::seqoff(seq_expt_, proto->m_ptk_seq) > 0 ){
		pkt_loss_ += NtdSeq::seqoff(seq_expt_, proto->m_ptk_seq);
		LOG4CPLUS_ERROR(g_logger,"pkt loss, "<<NtdSeq::seqoff(seq_expt_, proto->m_ptk_seq)<< " total_loss = "<<pkt_loss_);
		seq_expt_ = NtdSeq::incseq(proto->m_ptk_seq);
	}else if( NtdSeq::seqoff(seq_expt_, proto->m_ptk_seq) == 0){
		seq_expt_ = NtdSeq::incseq(proto->m_ptk_seq);
	}else{
		LOG4CPLUS_ERROR(g_logger, "disorder pkt, seq = "<<proto->m_ptk_seq <<" expect = "<< seq_expt_);
		if( pkt_loss_ > 0){
			--pkt_loss_;
		}
		//assert(false);
	}

	if(start_ts_ != last_io_ts_)//skip first packet
	{
		//assert(m_tscost);
		//assert(m_seqexpt!=m_initSeq);
		NtdProtoFeedback * feed_back = new NtdProtoFeedback();
		//feed_back->m_pkt_loss = pkt_loss_*100/NtdSeq::seqoff(seq_init_, seq_expt_);
		feed_back->m_pkt_loss = pkt_loss_;
		//m_feedback.m_recving_rate = m_sm.GetSpeed();
		feed_back->m_ptk_seq = proto->m_ptk_seq;
		feed_back->m_recving_rate = static_cast<uint32>(recv_total_bytes_*1000/life_time_);
		feed_back->m_timestamp_sending = proto->m_timestamp_sending;
		feed_back->m_timestamp_recving = ts;
		feed_back->m_convId = proto->m_convId;

		LOG4CPLUS_TRACE(g_logger," [server] - "<<feed_back->Desc());
		LOG4CPLUS_TRACE(g_logger,"recv: "<<feed_back->m_timestamp_recving<< " send: " << feed_back->m_timestamp_sending
			<<" sample: "<<static_cast<uint64>(feed_back->m_timestamp_recving - feed_back->m_timestamp_sending));

		TrySend(feed_back);
	}
}

void Connection::HandleProtoRawData(NtdProtoRawData * proto, void* pContext)
{
	//signal read event
	port_->OnReadPacket(this, (const char*)proto->m_pData, proto->m_len);
}

void Connection::HandleProtoFeedback(NtdProtoFeedback * proto, void* pContext)
{
	//add the samples

	uint32 now = Now_ms();
	uint32 rtt = now - proto->m_timestamp_sending;


	uint32 owd = proto->m_timestamp_recving - proto->m_timestamp_sending;
	uint32 reverse_owd = now - proto->m_timestamp_recving;

	//if(owd_base_ == 0){
	//	owd_base_ = owd;
	//}else{
	//	int32 owd_var = owd - owd_base_;

	//	if( smoothed_rtt_ == 0 ){
	//		smoothed_rtt_ = rtt;
	//	}else{
	//		smoothed_rtt_ = smoothed_rtt_ * 0.8 + rtt * 0.2;
	//	}

	//	//if( GetProtoType() == PROTO_UDP ){
	//	//	LOG4CPLUS_DEBUG(g_logger, "["<<GetProtoType()<<"]"<<" conv_id = "<< proto->m_convId 
	//	//		<< " now = "<< now <<" ms, send_ts = "<<proto->m_timestamp_sending
	//	//		<< " ms, rtt = " << rtt <<" ms, rel_owd = " << owd_var <<" ms "
	//	//		<<" recv_rate = " << proto->m_recving_rate << "Bytes/sec "
	//	//		<<" pkt_loss = " << proto->m_pkt_loss << "%");
	//	//}else{
	//	//	LOG4CPLUS_DEBUG(g_logger, "["<<GetProtoType()<<"]"<<" conv_id = "<< proto->m_convId 
	//	//		<< " now = "<< now <<" ms, send_ts = "<<proto->m_timestamp_sending
	//	//		<< " ms, rtt = " << rtt <<" ms, rel_owd = " << owd_var <<" ms "
	//	//		<<" recv_rate = " << proto->m_recving_rate << "Bytes/sec "
	//	//		<<" pkt_loss = " << proto->m_pkt_loss << "%");
	//	//}
	//}	

	NtdMetrics metrics;

	metrics.conv_id = proto->m_convId;
	metrics.seq = proto->m_ptk_seq;
	metrics.tick = proto->m_timestamp_sending;
	metrics.loss = proto->m_pkt_loss;
	metrics.reverse_owd = reverse_owd;
	metrics.owd = owd;
	metrics.rtt = rtt;
	metrics.recv_rate = proto->m_recving_rate;//TODO: Add speed meter
	metrics.send_rate = 0;//TODO: Add speed meter

	//#define PROTO_T0_STREAM_TYPE(proto_type) (proto_type==PROTO_TCP?NtdMetrics::STREAM_TCP:NtdMetrics::STREAM_UDP)
	metrics.proto_type = GetProtoType();

	AddSample(metrics);
}




//////////////////////////////////////////////////////////////////////////
//    callbacks
//////////////////////////////////////////////////////////////////////////

static void cb_func(evutil_socket_t fd, short what, void *user_data)
{
	if( what&EV_READ ){
		(static_cast<IOEventHandle *>(user_data))->OnRead();
	}

	if( what&EV_WRITE ){
		(static_cast<IOEventHandle *>(user_data))->OnWrite();
	}
}

//////////////////////////////////////////////////////////////////////////
//    ProxyConnection
//////////////////////////////////////////////////////////////////////////
ProxyConnection::ProxyConnection(Port* port, const SocketAddress& remote_addr)
: Connection(port, remote_addr)
{
	SetConnected(true);
    //seems not so good to callback in constructor
	//TODO: do we need handshake before notify connected?
	port_->OnConnected(this);
}

ProxyConnection::~ProxyConnection()
{ 
	port_->OnClose(this, 0); 
}

int ProxyConnection::DoSend(const void* data, size_t size) 
{
	int sent = port_->SendTo(data, size, remote_addr_);
	if (sent <= 0) {
		error_ = port_->GetError();
		if( !UTDUTIL_ERR_RW_RETRIABLE(error_) ){
			LOG4CPLUS_ERROR(g_logger, "sent failed, "<<remote_addr_.ToString());
			port_->OnError(this, error_);
		}
	} else {
		sent_total_bytes_ += sent;
	}
	return sent;
}

//////////////////////////////////////////////////////////////////////////
//    UDPPort
//////////////////////////////////////////////////////////////////////////


UDPPort::UDPPort(const SocketAddress& address)
{
	socket_ = socket( PF_INET, SOCK_DGRAM, IPPROTO_UDP );

	//int RcvBufferSize = 1024*1024;
	//setsockopt( socket_, SOL_SOCKET, SO_RCVBUF, (char*)&RcvBufferSize, sizeof( RcvBufferSize ) );
	SetOption(socket_, OPT_RCVBUF, 1024*1024);

	evutil_make_socket_nonblocking(socket_);

	read_ev_ = event_new(AsynEventFramework::Instance()->GetEventBase(),
		                 socket_, 
						 EV_READ|EV_PERSIST, 
						 cb_func,
						 dynamic_cast<IOEventHandle *>(this));
	event_add(read_ev_, NULL);

	write_ev_ = event_new(AsynEventFramework::Instance()->GetEventBase(),
		                  socket_, 
						  EV_WRITE, 
						  cb_func,
						  dynamic_cast<IOEventHandle *>(this));
	event_add(write_ev_, NULL);

	struct sockaddr_in sin;
	SetSockAddr(sin, address.ip(), address.port());
	if( bind( socket_, (sockaddr*)&sin, sizeof(sin) ) == -1 ){
		LOG4CPLUS_ERROR(g_logger, "bind error, "<<address.ToString());
	}
}


UDPPort::~UDPPort() 
{
	event_del(read_ev_);
	event_free(read_ev_);

	event_del(write_ev_);
	event_free(write_ev_);
}

Connection* UDPPort::CreateConnection(const SocketAddress& remote_addr) 
{
	Connection * conn = new ProxyConnection(this, remote_addr);
	conn->SetProtoType(Connection::PROTO_UDP);
	AddConnection(conn);
	return conn;
}

int UDPPort::SendTo(const void* data, size_t size, const SocketAddress& addr) 
{
	struct sockaddr_in dest_addr;
	SetSockAddr(dest_addr, addr.ip(), addr.port());

	int32 nSendBytes = sendto(socket_, (const char*)data, size, 0, (sockaddr *)&dest_addr, sizeof(dest_addr));

	if( nSendBytes < 0 ){
		error_ = evutil_socket_geterror(socket_);
		if(UTDUTIL_ERR_RW_RETRIABLE(error_) && !event_pending(write_ev_, EV_WRITE,NULL)){
			LOG4CPLUS_TRACE(g_logger,"add write_event notification");
			event_add(write_ev_, NULL);
			NotifyWriatable(false);
		}else{
			LOG4CPLUS_ERROR(g_logger, "sent to "<<addr.ToString()<<" error: "<<error_);
		}
	}

	return nSendBytes;
}

//int UDPPort::GetError()
//{
//	return evutil_socket_geterror(socket_);
//}

void UDPPort::OnReadPacket( const char* data, size_t size, const SocketAddress& remote_addr)
{
	Connection* conn = GetConnection(remote_addr);
	if (conn) {
		conn->OnReadPacket(data, size);
	} else {
		conn = CreateConnection(remote_addr);
		if (conn) {
			conn->OnReadPacket(data, size);
		} 
	}
}

void UDPPort::OnRead()
{
	int8 packet[TEMP_BUFFER];
	struct sockaddr_in addr;
	ev_socklen_t addrlen;
	int r;

	while(true) {
		addrlen = sizeof(struct sockaddr_in);
		r = recvfrom(socket_, packet, sizeof(packet), 0, (struct sockaddr*) &addr, &addrlen);
		SocketAddress sa;
		sa.FromSockAddr(addr);
		if (r < 0) {
			error_ = evutil_socket_geterror(socket_);
			if( !UTDUTIL_ERR_RW_RETRIABLE(error_) ){
				LOG4CPLUS_ERROR(g_logger, "Error while reading, "<<evutil_socket_error_to_string(error_));
				OnError(GetConnection(sa), error_);
			}
			return;
		}

		if( r == 0 ){
			LOG4CPLUS_ERROR(g_logger,"Incorrect UDP packet");
			return;
		}

		OnReadPacket(packet, r, sa);
	}
}

void UDPPort::NotifyWriatable(bool writable)
{
	//Signal write event
	AddressMap::iterator iter = connections_.begin();
	AddressMap::iterator end = connections_.end();
	for(; iter!=end; ++iter){
		//bool ret = iter->second->Flush();
		//if( ret ) {
		//	//Signal write event
		//	OnWritable(iter->second, writable);
		//}

		//TODO: we'd better avoid re-entry problem by posting the events
		OnWritable(iter->second, writable);
	}
}

void UDPPort::OnWrite()
{
	LOG4CPLUS_TRACE(g_logger,"remove write_event notification");

	event_del(write_ev_);

	bool ret = true;
	AddressMap::iterator iter = connections_.begin();
	AddressMap::iterator end = connections_.end();
	for(; iter!=end; ++iter){
		ret = iter->second->Flush();
		if( !ret ) {
			break;
		}
	}

	if( ret ){
		NotifyWriatable(true);
	}
}

//////////////////////////////////////////////////////////////////////////
//    TCPPort
//////////////////////////////////////////////////////////////////////////

TCPPort::TCPPort(const SocketAddress& address)
:address_(address)
,incoming_only_(address.port() != 0)
{
	socket_ = socket( PF_INET, SOCK_STREAM, IPPROTO_TCP );

	//int RcvBufferSize = 1024*1024;
	//setsockopt( socket_, SOL_SOCKET, SO_RCVBUF, (char*)&RcvBufferSize, sizeof( RcvBufferSize ) );
	//SetOption(socket_, OPT_RCVBUF, 1024*1024);

	evutil_make_socket_nonblocking(socket_);

	listen_ev_ = event_new(AsynEventFramework::Instance()->GetEventBase(),
		socket_, 
		EV_READ|EV_PERSIST, 
		cb_func,
		dynamic_cast<IOEventHandle *>(this));

	event_add(listen_ev_, NULL);

	struct sockaddr_in sin;
	SetSockAddr(sin, address.ip(), address.port());
	if( bind( socket_, (sockaddr*)&sin, sizeof(sin) ) == -1 ){
		LOG4CPLUS_ERROR(g_logger, "bind error, "<<address.ToString());
	}

	if( ::listen(socket_, 5) < 0 ){
		LOG4CPLUS_ERROR(g_logger, "Listen failed on" << address.ToString());
	}

	//socket_->SignalReadEvent.connect(this, &TCPPort::OnAcceptEvent);
}

TCPPort::~TCPPort() {
	event_del(listen_ev_);
	event_free(listen_ev_);
}

Connection* TCPPort::CreateConnection(const SocketAddress& remote_addr) 
{
	TCPConnection* conn = 0;
	if (evutil_socket_t socket = GetIncoming(remote_addr, true)) {
		//socket->SignalReadPacket.disconnect(this);
		conn = new TCPConnection(this, remote_addr, socket);
	} else {
	// Check if we are allowed to make outgoing TCP connections
		if (incoming_only_){
			return NULL;
		}else{
			conn = new TCPConnection(this, remote_addr);
		}
	}
	AddConnection(conn);
	return conn;
}

int TCPPort::SendTo(const void* data, size_t size, const SocketAddress& addr)
{
	evutil_socket_t socket = 0;

	if (TCPConnection * conn = static_cast<TCPConnection*>(GetConnection(addr))) {
		socket = conn->socket();
	} else {
		socket = GetIncoming(addr);
	}
	if (!socket) {
		LOG4CPLUS_ERROR(g_logger, "Unknown destination: " << addr.ToString());
		return -1; 
	}

	int sent = send(socket, (const char*)data, size, 0);
	if (sent < 0) {
		LOG4CPLUS_ERROR(g_logger, "send error: " << evutil_socket_geterror(socket));
	}
	return sent;
}

int TCPPort::GetError() {
	return evutil_socket_geterror(socket_);
}

void TCPPort::OnRead() 
{
	OnAcceptEvent();
}

void TCPPort::OnAcceptEvent() 
{
	while (true) {
		struct sockaddr_in sa;
		ev_socklen_t socklen = sizeof(sa);

		Incoming incoming;
		incoming.socket = ::accept(socket_, (sockaddr *)&sa, &socklen);
		if (incoming.socket < 0)
			break;

		evutil_make_socket_nonblocking(incoming.socket);
        SetOption(socket_, OPT_RCVBUF, 1024*1024);
		//int ret = SetOption(socket_, OPT_NODELAY, 1);

		incoming.addr.FromSockAddr(sa);

		incoming_.push_back(incoming);

		CreateConnection(incoming.addr);

		LOG4CPLUS_DEBUG(g_logger, "Accept from "<< "(" << incoming.addr.ToString() << ")");
	}

	error_ = evutil_socket_geterror(socket_);
	if (UTDUTIL_ERR_RW_RETRIABLE(error_))
		return;

	LOG4CPLUS_ERROR(g_logger, "Error from accept() call: "<< error_);
}

evutil_socket_t TCPPort::GetIncoming(const SocketAddress& addr, bool remove) 
{
	evutil_socket_t socket = 0;
	for (std::list<Incoming>::iterator it = incoming_.begin();
		it != incoming_.end(); ++it)
	{
		if (it->addr == addr) {
			socket = it->socket;
			if (remove)
				incoming_.erase(it);
			break;
		}
	}
	return socket;
}

//void TCPPort::OnReadPacket(const char* data, size_t size, 
//						   const SocketAddress& remote_addr) 
//{
//	Port::OnReadPacket(data, size, remote_addr);
//}

TCPConnection::TCPConnection(TCPPort* port, const SocketAddress& remote_addr, evutil_socket_t socket)
: Connection(port, remote_addr), socket_(socket)
{
	proto_type_ = PROTO_TCP;

	bool outgoing = (socket_ == INVALID_SOCKET);
	if (outgoing) {
		socket_ = ::socket( PF_INET, SOCK_STREAM, IPPROTO_TCP );
		int ret = SetOption(socket_, OPT_NODELAY, 1);
	} 

	read_ev_ = event_new(AsynEventFramework::Instance()->GetEventBase(),
		socket_, 
		EV_READ|EV_PERSIST, 
		cb_func,
		dynamic_cast<IOEventHandle *>(this));
	event_add(read_ev_, NULL);

	write_ev_ = event_new(AsynEventFramework::Instance()->GetEventBase(),
		socket_, 
		EV_WRITE, 
		cb_func,
		dynamic_cast<IOEventHandle *>(this));
	event_add(write_ev_, NULL);

	if (outgoing) {
		SetConnected(false);
		struct sockaddr_in sin;
		SetSockAddr(sin, port->address_.ip(), 0);
		if( bind( socket_, (sockaddr*)&sin, sizeof(sin) ) == -1 ){
			LOG4CPLUS_ERROR(g_logger, "bind error, "<<port->address_.ToString());
		}

		evutil_make_socket_nonblocking(socket_);

		SetSockAddr(sin, remote_addr.ip(), remote_addr.port());

		if (connect(socket_, (sockaddr*)&sin, sizeof(sockaddr)) < 0) {
			int e = evutil_socket_geterror(socket_);
			if (UTDUTIL_ERR_CONNECT_RETRIABLE(e)){
				LOG4CPLUS_DEBUG(g_logger, "connecting ... " << remote_addr.ToString());
			}
			if (UTDUTIL_ERR_CONNECT_REFUSED(e)){
				LOG4CPLUS_ERROR(g_logger, "connecting refused!");
			}
		} else {
			LOG4CPLUS_DEBUG(g_logger, "connecting succeed, " << remote_addr.ToString());
		}
	}else{
		//seems not so good to callback in constructor
		port_->OnConnected(this);
	}
}

TCPConnection::~TCPConnection() {
	evutil_closesocket(socket_);

	event_del(read_ev_);
	event_free(read_ev_);

	event_del(write_ev_);
	event_free(write_ev_);
}

int TCPConnection::DoSend(const void* data, size_t size) {
	//if (write_state() != STATE_WRITABLE) {
	//	// TODO: Should STATE_WRITE_TIMEOUT return a non-blocking error?
	//	error_ = EWOULDBLOCK;
	//	return SOCKET_ERROR;
	//}

	//TODO: encode in RawData first
	
	size_t  total = 0;
	while( total < size ){
		int sent = ::send(socket_, (const char *)data, size, 0);

		if( sent < 0 ){
			int err = evutil_socket_geterror(socket_);
			if(UTDUTIL_ERR_RW_RETRIABLE(err) && !event_pending(write_ev_, EV_WRITE,NULL)){
				LOG4CPLUS_TRACE(g_logger,"add write_event notification");
				event_add(write_ev_, NULL);
				port_->OnWritable(this, false);
			}else{
				LOG4CPLUS_ERROR(g_logger, "sent to "<<remote_addr_.ToString()<<" error: "<<err);
				port_->OnError(this, err);
			}
			break;
		}else{
			sent_total_bytes_ += sent;			
			total += sent;
		}
  
	}

	return total;
}

void TCPConnection::OnConnect() 
{
	SetConnected(true);
	port_->OnConnected(this);
	LOG4CPLUS_ERROR(g_logger, "connected to " << remote_addr_.ToString());
}

void TCPConnection::OnClose() 
{
	//signal close event
	SetConnected(false);
	port_->OnClose(this, GetError());
	LOG4CPLUS_ERROR(g_logger, "OnClose " << remote_addr_.ToString() << " err: "<<GetError());

	//should delete here?
	//delete this;
	//Dispose(this);
}

//void TCPConnection::OnReadPacket(const char* data, size_t size) 
//{
//	Connection::OnReadPacket(data, size);
//}

void TCPConnection::OnRead()
{
	int8 packet[TEMP_BUFFER];
	int r;

	while(true) {
		r = recv(socket_, packet, sizeof(packet), 0);
		if (r < 0) {
			error_ = evutil_socket_geterror(socket_);
			if( !UTDUTIL_ERR_RW_RETRIABLE(error_) ){
				LOG4CPLUS_ERROR(g_logger,"Error while reading, "<<evutil_socket_error_to_string(error_));
				port_->OnError(this, error_);
				OnClose();
			}
			return;
		}

		if( r == 0 ){
			LOG4CPLUS_ERROR(g_logger,"Socket close");
			port_->OnError(this, 0);
			OnClose();
			return;
		}

		OnReadPacket(packet, r);
	}
}

void TCPConnection::OnWrite()
{
	LOG4CPLUS_TRACE(g_logger,"remove write_event notification");
	event_del(write_ev_);

	if( !connected_ ){
		OnConnect();
	}

	//Resent left bytes
	//TODO: do we need to take care of the return value?
	//SendOutputbuf();
	bool ret = Flush();

	if( ret ) {
		//Signal write event
		port_->OnWritable(this, true);
	}
}

//void TCPConnection::OnReadPacket(const char* data, size_t size)
//{
//	Connection::OnReadPacket(data, size);
//}

//////////////////////////////////////////////////////////////////////////
ProbingManager::ProbingManager()
{
	precision_tick_ = PRECISION_TICK;//ms
	interval_ = PROBING_DELAY;
	notifying_ = false;
	SetPrecision(precision_tick_);
	SetProbingInterval(interval_);

    probe_timer_.AddListener(this);
}

ProbingManager::~ProbingManager()
{
	probe_timer_.Stop();
	probe_timer_.RemoveListener(this);
}

void ProbingManager::AddConnection(IConnection * pConn)
{
	conn_set_.insert(pConn);
	SetProbingInterval(pConn, interval_);
}

void ProbingManager::RemoveConnection(IConnection * pConn)
{
	conn_set_.erase(pConn);
}

void ProbingManager::SetPrecision(int ms)
{
	if( ms == INT_MAX ){
		//stop probing timer
		precision_tick_ = ms;
		probe_timer_.Stop();
	}else{
		// if( ms != interval_ )
		precision_tick_ = ms;
		probe_timer_.Start(precision_tick_, true);
	}
}

void ProbingManager::SetProbingInterval(int ms)
{
	interval_ = ms;
    std::for_each(conn_set_.begin(), conn_set_.end(), std::bind2nd(std::mem_fun1(&IConnection::SetProbingInterval), ms));
}

void ProbingManager::SetProbingInterval(IConnection * pConn, int ms)
{
	assert(pConn);
	pConn->SetProbingInterval(ms);
}

//This will be called whenever the connection itself trigger the ping
//Other Connections will check if they need to ping now
void ProbingManager::OnTick()
{
	//avoid re-entry problem, not so good, but simple, need better design
	if(notifying_) return;

	notifying_ = true;
	OnTimer(NULL);
	notifying_ = false;
}

void ProbingManager::OnTimer(NTDTimer* pTimer)
{
	uint32 now = Now_ms();

	std::set<IConnection *>::iterator iter = conn_set_.begin();
	std::set<IConnection *>::iterator end = conn_set_.end();
	for(; iter!=end; ++iter){
		PingIfNeeded(now, (*iter));
	}
}

void ProbingManager::PingIfNeeded(uint32 now, IConnection * pConn)
{
	if( pConn->IsTimetoPing(now) && pConn->Connected() ){
		pConn->DoPing();
	}
}